package com.shengzai.flink.sink

import org.apache.flink.api.common.functions.RuntimeContext
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}

import java.sql.{Connection, DriverManager, PreparedStatement}

object Demo4MySQLSink {

  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    val fileDS: DataStream[String] = env.readTextFile("data/students.txt")

    fileDS.addSink(new mysqlSink)

    env.execute()


  }

}
/**
 * 自定义sink将数据写入mysql
 * RichSinkFunction比SinkFunction多了open和close方法，用于创建和关闭连接
 */
class mysqlSink extends RichSinkFunction[String]{

  var conn :Connection= _
  var stat :PreparedStatement =_

  override def open(parameters: Configuration): Unit = {
    Class.forName("com.mysql.jdbc.Driver")
    conn = DriverManager.getConnection("jdbc:mysql://master:3306/bigdata?useUnicode=true&characterEncoding=UTF-8", "root", "123456")
    stat = conn.prepareStatement("insert into Teacher values (?,?)")
  }

  override def close(): Unit ={
    conn.close()
  }

  override def invoke(value: String, context: SinkFunction.Context): Unit ={
    stat.setInt(1,5)
    stat.setString(2,"王五")
    stat.execute()
  }
}



